package com.cmnit.model

import java.util.Properties

import com.cmnit.utils.{ConfigurationManager, DataSourceUtil, DateUtil, ModelUtils, MySqlUtils}
import org.apache.log4j.Logger
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}

object TollLicencePlate {
  val logger: Logger = Logger.getLogger(TollLicencePlate.getClass)

  def run(sparkSession: SparkSession, stationFrameBroadCast: Broadcast[DataFrame], vehicleBasicBroadCast: Broadcast[DataFrame], year: String, month: String, day: String, hour: String): Unit = {

    // 获取加工账期（默认当前时间的上1小时）
    val (acctHour, acctDay, acctMonth, acctYear) = ModelUtils.getAcctDate(year, month, day, hour)

    // 拼接当前账期
    val acctTime = acctYear + "-" + acctMonth + "-" + acctDay + " " + acctHour + ":" + "00:00"
    logger.info("当前账期时间：" + acctTime)
    val acctDate = acctYear + acctMonth + acctDay
    // 得到当前账期date对象，计算前一个账期的时间
    val (lastAcctHour, lastAcctDay, lastAcctMonth, lastAcctYear) = ModelUtils.getTime(DateUtil.getTimeByString(acctTime))
    // 拼接前一个账期
    val lastAcctTime = lastAcctYear + "-" + lastAcctMonth + "-" + lastAcctDay + " " + lastAcctHour + ":" + "00:00"
    logger.info("前一个账期时间：" + lastAcctTime)

    stationFrameBroadCast.value.createOrReplaceTempView("tmp_organization_toll")
    logger.info("生成收费站中间表:tmp_organization_toll")
    vehicleBasicBroadCast.value.createOrReplaceTempView("tmp_tbl_vehiclebasicinfo")
    logger.info("生成车牌-车辆类型码表:tmp_tbl_vehiclebasicinfo")

    // 获取最新的收费站牌识数据
    sparkSession.sql(
      "select " +
        "* " +
        "from " +
        "ods.ods_etc_tollvehdisdata " +
        "where identifyVlp not like '%挂' " +
        "and hour= " + acctHour + " " +
        "and day=" + acctDay + " " +
        "and month=" + acctMonth + " " +
        "and year=" + acctYear)
      .createOrReplaceTempView("temp_etc_tollvehdisdata")
    logger.info("生成收费站出口表:temp_etc_tollvehdisdata")

    // 收费站牌识：dwd.dwd_traffic_tollvehdisdata_hi
    sparkSession.sql(
      " select  " +
        "t.id," +
        "'江西省' as province," +
        "t.TollStationId," +
        "tmp.tollstationName," +
        "tmp.roadId," +
        "tmp.roadName," +
        "t.laneNum," +
        "t.identifyVlp," +
        "t.vehicleId " +
        "from " +
        "temp_etc_tollvehdisdata t " +
        "left join dim.dim_tbl_gbsbstationbaseinfo d " +
        "on d.gbtollstationid = t.TollStationId " +
        "left join tmp_organization_toll tmp " +
        "on tmp.tollstationProvince = d.sbtollstationid ")
      .createOrReplaceTempView("tmp_tolldisdata_dwd")
    logger.info("生成收费站牌识表:tmp_tolldisdata_dwd")

    // 收费站流量：dws.dws_traffic_station_tollvecountdisdata_hi
    sparkSession.sql(
      "select " +
        "t.province," +
        "t.roadid," +
        "t.roadname," +
        "t.tollstationid," +
        "t.tollstationname," +
        "t.lanetype," +
        "t.vehicletype," +
        "count(t.vehicleid) as countnum " +
        "from " +
        "(select " +
        "a.*," +
        "nvl(b.vehicletype, 1) vehicletype " +
        "from " +
        "(select " +
        "distinct " +
        "vehicleid," +
        "province," +
        "roadid," +
        "roadname," +
        "tollstationid," +
        "tollstationname," +
        "vehicleid," +
        "if(lanenum < 100, '入口', '出口') as lanetype " +
        "from " +
        "tmp_tolldisdata_dwd) a " +
        "left join " +
        "tmp_tbl_vehiclebasicinfo b " +
        "on a.vehicleid = b.vehicleplate) t " +
        "group by " +
        "t.province,t.roadid,t.roadname,t.tollstationname,t.tollstationid,t.lanetype,t.vehicletype")
      .createOrReplaceTempView("tmp_tolldisdata_dws")
    logger.info("生成收费站牌识流水车流量表:tmp_tolldisdata_dws")

    val username = ConfigurationManager.getProperty("mysql.username")
    val password = ConfigurationManager.getProperty("mysql.password")
    val url = ConfigurationManager.getProperty("mysql.url")
    val driver = ConfigurationManager.getProperty("mysql.driver")
    val prop = new Properties
    prop.setProperty("user", username)
    prop.setProperty("password", password)
    prop.setProperty("driver", driver)

    sparkSession.read
      .jdbc(url, "traffic_province_lanevehicletypevecountdisdata_day".toLowerCase, prop)
      .createOrReplaceTempView("tmp_traffic_province_lanevehicletypevecountdisdata_day")
    logger.info("获取mysql结果表:tmp_traffic_province_lanevehicletypevecountdisdata_day")

    sparkSession.sql(
      "select " +
        "province," +
        "roadid," +
        "roadname," +
        "tollstationid," +
        "tollstationname," +
        "lanetype," +
        "vehicletype," +
        "countnum " +
        "from " +
        "tmp_traffic_province_lanevehicletypevecountdisdata_day " +
        "where " +
        "date = '" + acctDate + "' " +
        "and time = '" + lastAcctTime + "' "
    )
      .createOrReplaceTempView("tmp_mysql_toll")
    logger.info("获取mysql结果表的存量数据:tmp_mysql_toll")

    // 删除当账期的原数据
    val sql = "delete from traffic_province_lanevehicletypevecountdisdata_day where time = '" + acctTime + "' and date = '" + acctDate + "'"
    MySqlUtils.execute(sql, DataSourceUtil.getConnection)

    // 省界收费站出入省车流量，牌识计算
    sparkSession.sql(
      "select " +
        "nvl(a.province, b.province) province, " +
        "nvl(a.roadid, b.roadid) roadid, " +
        "nvl(a.roadname, b.roadname) roadname, " +
        "nvl(a.tollstationid, b.tollstationid) tollstationid, " +
        "nvl(a.tollstationname, b.tollstationname) tollstationname, " +
        "nvl(a.lanetype, b.lanetype) lanetype, " +
        "nvl(a.vehicletype, b.vehicletype) vehicletype, " +
        "nvl(a.countnum, 0) + nvl(b.countnum, 0) countnum, " +
        "'" + acctTime + "' as time, " +
        "'" + acctDate + "' as date " +
        "from " +
        "tmp_tolldisdata_dws a " +
        "full join " +
        "tmp_mysql_toll b " +
        "on " +
        "a.province = b.province " +
        "and a.roadid = b.roadid " +
        "and a.roadname = b.roadname " +
        "and a.tollstationid = b.tollstationid " +
        "and a.tollstationname = b.tollstationname " +
        "and a.lanetype = b.lanetype " +
        "and a.vehicletype = b.vehicletype ")
      .write
      .mode(SaveMode.Append)
      .jdbc(url, "traffic_province_lanevehicletypevecountdisdata_day".toLowerCase, prop)

    logger.info("生成省界收费站出入省车流量:traffic_province_lanevehicletypevecountdisdata_day")
    logger.info("插入mysql完毕")
    sparkSession.close
  }
}
